package app.dwm;

import app.dwd.BaseAppFollow;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONAware;
import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Lists;
import common.Constant;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import util.GmallSinkUtil;

import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.Date;

public class DWMUVAppFollow extends BaseAppFollow {

    public static void main(String[] args) {

        new DWMUVAppFollow().init(4001, 1, "DWMUVAppFollow", Constant.TOPIC_DWD_PAGE_LOG, "DWMUVAppFollow");
    }

    @Override
    public void run(StreamExecutionEnvironment env, DataStreamSource<String> dataStreamSource) {
        /*1 mid分组
         * 2 开窗,小窗时效性高,计算压力大;大窗时效性差,计算压力小
         * 3 通过状态是否为空判别是不是第一次窗内事件,第一个事件时间更新至状态中,将第一条数据发出至DWM层,留待DWS层求UV,其余数据舍去
         * 4 UV按天统计,所以需要第二天清空状态;如何确定第二天?一旦窗内元素和状态时间不相等(天)就说明是第二天,将状态清空;
         * 注意开窗后就无法使用定时器,不然这里还可以使用定时器的逻辑*/
        SingleOutputStreamOperator<JSONObject> midFirstEventEveryDay = getMidFirstEventEveryDay(dataStreamSource);

        GmallSinkUtil.sendToKafka(midFirstEventEveryDay.map(JSONAware::toJSONString), Constant.TOPIC_DWM_UV);

    }

    private SingleOutputStreamOperator<JSONObject> getMidFirstEventEveryDay(DataStreamSource<String> dataStreamSource) {
        return dataStreamSource
            .map(JSON::parseObject)
            .assignTimestampsAndWatermarks(WatermarkStrategy
                                               .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                               .withTimestampAssigner((obj, ts) -> obj.getLong("ts")))
            .keyBy(event -> event.getJSONObject("common").getString("mid"))
            .window(TumblingEventTimeWindows.of(Time.seconds(5)))
            .process(new ProcessWindowFunction<JSONObject, JSONObject, String, TimeWindow>() {

                private SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd");
                private ValueState<Long> firstTimeState;

                @Override
                public void process(String s,
                                    Context ctx,
                                    Iterable<JSONObject> eles,
                                    Collector<JSONObject> out) throws Exception {

                    if (firstTimeState.value() != null) {
                        String tomorrow = format.format(new Date(ctx.window().getEnd()));
                        String today = format.format(new Date(firstTimeState.value()));
                        if (!today.equals(tomorrow)) {
                            firstTimeState.clear();
                        }
                    }

                    if (firstTimeState.value() == null) {
                        ArrayList<JSONObject> list = Lists.newArrayList(eles);
                        Collections.min(list, new Comparator<JSONObject>() {
                            @Override
                            public int compare(JSONObject o1, JSONObject o2) {
                                return o1.getLong("ts").compareTo(o2.getLong("ts"));
                            }
                        });
                        JSONObject firstEvent = list.get(0);
                        out.collect(firstEvent);
                        firstTimeState.update(firstEvent.getLong("ts"));
                    }
                }

                @Override
                public void open(Configuration parameters) throws Exception {
                    firstTimeState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("firstTimeState", Long.class));
                }
            });
    }
}
